package com.zheng;

import com.zheng.client.RPCConsumer;
import com.zheng.service.UserService;
import com.zheng.zk.Curator;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;

import java.beans.Customizer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: zheng
 * @Date: 2020-06-28 23:02
 * @Description:
 */
public class CustomerBootStrap {
    public static CuratorFramework curator = Curator.getCuratorFramework();
    public static ConcurrentHashMap<String, RPCConsumer> providerMap = new ConcurrentHashMap<>();

    public static void main(String[] args) throws Exception {
        connectionZk(null);
        listenerZK();
        rpcStart();
        rpcStart2();
    }

    private static void listenerZK() throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curator, "/", false);
        PathChildrenCacheListener listener = new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                switch (pathChildrenCacheEvent.getType()) {
                    case CHILD_ADDED:
                        System.out.println("添加节点" + pathChildrenCacheEvent.getData());
                        connectionZk(null);
                        break;
                    case CHILD_REMOVED:
                        System.out.println("移除节点" + pathChildrenCacheEvent.getData());
                        providerMap.remove(pathChildrenCacheEvent.getData().getPath());
                        connectionZk(null);
                        break;
                }
            }
        };
        pathChildrenCache.getListenable().addListener(listener);
        pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
    }

    public static void connectionZk(Map map) throws Exception {
        //获取base节点下面的子节点
        List<String> zkList = curator.getChildren().forPath("/");
        for (String sr : zkList) {
            //获取每个节点下面的内容
            byte[] bytes = curator.getData().forPath("/" + sr);
            String data = new String(bytes);
            String[] datas = data.split(":");
            String host = datas[1];
            String port = datas[3];
            String time = datas[4];
            RPCConsumer rpcConsumer = null;
            //重制标志位
            if (map == null) {
                rpcConsumer = new RPCConsumer(host, Integer.parseInt(port), Long.parseLong(time), sr);
            } else {
                //启动重制,时间为0
                Integer restPort = (Integer) map.get("port");
                if (restPort == Integer.parseInt(port)) {
                    rpcConsumer = new RPCConsumer(host, Integer.parseInt(port), Long.parseLong("0"), sr);
                    String updateData = "host:" + host + ":port:" + port + ":0";
                    curator.setData().forPath("/" + sr, updateData.getBytes());
                } else {
                    rpcConsumer = new RPCConsumer(host, Integer.parseInt(port), Long.parseLong(time), sr);
                }

            }
            providerMap.put("/" + sr, rpcConsumer);
        }
    }

    public static void rpcStart() throws InterruptedException {
        while (true) {
            for (Map.Entry<String, RPCConsumer> entry : providerMap.entrySet()) {
                RPCConsumer rpcConsumer = entry.getValue();
                System.out.println(rpcConsumer.getPort());
                UserService userService = (UserService) rpcConsumer.createProxy(UserService.class);
                userService.sayHello("RPC开始建立连接");
                System.out.println("使用连接为" + rpcConsumer.getIp() + "端口为" + rpcConsumer.getPort());
                Thread.sleep(5000);
            }
        }
    }

    //按照时间进行负载均衡
    public static void rpcStart2() throws Exception {
        ConcurrentHashMap<Integer, Long> resetMap = new ConcurrentHashMap<>();
        boolean flag = true;
        while (true) {
            List<RPCConsumer> list = new ArrayList<>();
            for (Map.Entry<String, RPCConsumer> entry : providerMap.entrySet()) {
                list.add(entry.getValue());
            }
            //按照时间进行排序
            Collections.sort(list, new Comparator<RPCConsumer>() {
                @Override
                public int compare(RPCConsumer o1, RPCConsumer o2) {
                    if (o1.getTime() > o2.getTime()) {
                        return 1;
                    } else if (o1.getTime() == o2.getTime()) {
                        return 0;
                    }
                    return -1;
                }
            });
            if (list.size() > 0) {
                    RPCConsumer rpcConsumer = list.get(0);
                    UserService userService = (UserService) rpcConsumer.createProxy(UserService.class);
                    userService.sayHello("RPC开始建立连接"+rpcConsumer.getPort());
                    System.out.println("使用连接为" + rpcConsumer.getIp() + "端口为" + rpcConsumer.getPort());
                    Thread.sleep(5000);
                    resetMap.put(rpcConsumer.getPort(), System.currentTimeMillis());
                    System.out.println("端口" + rpcConsumer.getPort() + "==" + System.currentTimeMillis());


            }
            //记录剩余没有进行调用的服务端口,因为list已经排序过，所以没有记录的服务端口应该从下标1开始记录，并赋值为0,表示没启动
            if (list.size() > 1) {
                for (int i = 1; i < list.size(); i++) {
                    if (resetMap.get(list.get(i).getPort()) == null) {
                        resetMap.put(list.get(i).getPort(), 0l);
                    }

                }
            }
            for (Map.Entry<Integer, Long> entry : resetMap.entrySet()) {
                long nowTime = System.currentTimeMillis();
//                  System.out.println("当前时间"+nowTime);
                long restTime = nowTime - entry.getValue();
//                  System.out.println("port"+entry.getKey()+"time"+entry.getValue());
                  System.out.println("差值"+restTime);
                if (restTime > 25000) {
                    //重新获取最新的数据并重制5s内没有连接的机器
                    Map map = new HashMap();
                    map.put("port", entry.getKey());
                    connectionZk(map);
                    flag = false;
                    break;
                }
            }

            if (flag) {
                //重新获取最新的数据
                connectionZk(null);

            }
            flag=true;


        }
    }


}
